package com.github.baichuan.iot.gateway.handler;

import com.github.baichuan.iot.protocol.codec.MessageCodec;
import com.github.baichuan.iot.protocol.codec.MessageDecodeContext;
import com.github.baichuan.iot.protocol.codec.MessageEncodeContext;
import com.github.baichuan.iot.protocol.codec.MessageException;
import io.netty.buffer.ByteBuf;
import io.netty.buffer.ByteBufUtil;
import io.netty.channel.ChannelDuplexHandler;
import io.netty.channel.ChannelHandler;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.ChannelPromise;
import io.netty.channel.socket.DatagramPacket;
import io.netty.util.ReferenceCountUtil;
import lombok.extern.slf4j.Slf4j;

import java.net.SocketAddress;
import java.time.LocalDateTime;

/**
 * 协议消息编解码处理器
 */
@Slf4j
@ChannelHandler.Sharable
public class MessageCodecHandler extends ChannelDuplexHandler {

    private MessageCodec codec;

    public MessageCodecHandler(MessageCodec codec){
        this.codec = codec;
    }

    @Override
    public void channelRead(ChannelHandlerContext ctx, Object msg){
        ByteBuf buf = null;
        SocketAddress address = null;
        try {
            if(msg instanceof ByteBuf){
                // TCP
                buf = (ByteBuf) msg;
                address = ctx.channel().remoteAddress();
            } else if(msg instanceof DatagramPacket){
                // UDP
                DatagramPacket packet = (DatagramPacket) msg;
                buf = packet.content().copy();
                address = packet.sender();
            } else{
                log.warn("can not decode msg class {}", msg.getClass().getName());
                return;
            }

            log.trace("Inbound datagram {} from {}", ByteBufUtil.hexDump(buf).toUpperCase(), address);

            MessageDecodeContext inbound = codec.decode(buf);
            inbound.setIngressTime(LocalDateTime.now());

            ctx.fireChannelRead(inbound);
        } catch (MessageException e) {
            log.error(e.getMessage(), e);
        } finally {
            ReferenceCountUtil.safeRelease(buf);
        }
    }

    @Override
    public void write(ChannelHandlerContext ctx, Object msg, ChannelPromise promise) throws Exception {
        if(!(msg instanceof MessageEncodeContext)){
            log.error("Not support outbound message class {}", msg.getClass().getName());
            return;
        }

        MessageEncodeContext outbound = (MessageEncodeContext) msg;
        ByteBuf buf = codec.encode(outbound);
        outbound.setDatagram(ByteBufUtil.getBytes(buf));
        super.write(ctx, outbound, promise);
    }
}

